#include "segment/segment.h"

namespace hawking {
namespace indexlib {

DEFINE_uint32(max_fetch_doc_count, 10000, "max docs fetch from inverted index");
DEFINE_int32(tomb_init_size, 1024, "tomb init size");
DECLARE_string(default_ttl_col_name);

void Segment::MoveToResponse_(
    SegmentContext* segment_ctx,
    const google::protobuf::RepeatedPtrField<std::string>& select_fields,
    IndexlibRetrieveResponse* resp) const {
    std::vector<DocContext>& doc_context = *(segment_ctx->MutableDocContext());
    for (size_t idx = 0; idx < doc_context.size(); ++idx) {
        if (doc_context[idx].filtered) {
            continue;
        }

        IndexlibDocRespones* doc = resp->add_docs();
        for (const auto& select_field : select_fields) {
            util::ColumnType field_type;
            if (!GetFieldIdAndTypedByName(select_field, &field_id, &field_type)) {
                continue;
            }

            util::FieldValue* field_value = doc_context[idx].GetField(field_id);
            if (field_value) {
                util::SelectColumn* new_col = doc->add_result();
                new_col->set_column_name(select_field);
                new_col->mutable_column_value()->Swap(field_value);
            } else {
                if (field_type == util::ColumnType::COLUMN_INT8 ||
                    field_type == util::ColumnType::COLUMN_INT16 ||
                    field_type == util::ColumnType::COLUMN_INT32 ||
                    field_type == util::ColumnType::COLUMN_INT64) {
                    auto iter = doc_context[idx].single_int_field.find(field_id);
                    if (iter != doc_context[idx].single_int_field.end()) {
                        util::SelectColumn* new_col = doc->add_result();
                        new_col->set_column_name(select_field);
                        new_col->mutable_column_value()->set_int_value(iter->second);
                    }
                } else if (field_type == util::ColumnType::COLUMN_FLOAT) {
                    auto iter = doc_context[idx].single_float_field.find(field_id);
                    if (iter != doc_context[idx].single_float_field.end()) {
                        util::SelectColumn* new_col = doc->add_result();
                        new_col->set_column_name(select_field);
                        new_col->mutable_column_value()->set_double_value(iter->second);
                    }
                } else if (field_type == util::ColumnType::COLUMN_DOUBLE) {
                    auto iter = doc_context[idx].single_double_field.find(field_id);
                    if (iter != doc_context[idx].single_double_field.end()) {
                        util::SelectColumn* new_col = doc->add_result();
                        new_col->set_column_name(select_field);
                        new_col->mutable_column_value()->set_double_value(iter->second);
                    }
                }
            }
        }
    }
}

void MemSegment::ObsoletedFilter_(
    std::unordered_map<DocId, DocId>* old2new_docid_map) const {
    if (!TTL()) {
        return;
    }

    int32_t field_id;
    util::ColumnType field_type;
    if (!GetFieldIdAndTypedByName(FLAGS_default_ttl_col_name, &field_id, &field_type)) {
        LOG(WARNING) << "Get field " << FLAGS_default_ttl_col_name << " fail";
        return;
    }

    std::shared_ptr<AttributeIndexer> ttl_indexer = GetAttributeIndexer(field_id);
    if (!ttl_indexer) {
        LOG(WARNING) << "attr index of  field "
            << FLAGS_default_ttl_col_name << " is nullptr";
        return;
    }
    
    auto pk_doc_map_iter = pk_indexer_->CreateIterator();       

    int64_t current_ts = butil::gettimeofday_s();
    uint64_t pk;
    DocId doc_id;
    indexlib::ArenaWrapper arena_wrapper;
    SegmentContext segment_ctx(arena_wrapper);
    std::unordered_set<DocId> fileter_doc_ids;
    while (pk_doc_map_iter->HasNext()) {
        pk_doc_map_iter->Next();
        pk_doc_map_iter->Value(&pk, &doc_id);

        segment_ctx.PushDocId(doc_id);
        ttl_indexer->BatchRead(&segment_ctx);
        auto iter = segment_ctx->DocContexts().single_int_field.find(field_id);
        if (iter != segment_ctx->DocContexts().single_int_field.end() &&
            iter->second <= current_ts) {
            fileter_doc_ids.insert(doc_id);
        }

        segment_ctx.Clear();
    }

    if (!fileter_doc_ids.empty()) {
        DocId new_doc_id = 0;
        std::unordered_map<DocId, DocId> new_old2new_docid_map;
        new_old2new_docid_map.reserve(old2new_docid_map.size());
        for (const auto& iter : *old2new_docid_map) {
            if (fileter_doc_ids.find(iter->first) == fileter_doc_ids.end()) {
                new_old2new_docid_map.emplace(iter->first, new_doc_id++);
            }
        }

        old2new_docid_map->swap(new_old2new_docid_map);
    }
}

Status Segment::ReadData_(
    std::shared_ptr<AttributeIndexer> attr_indexer,
    SegmentContext* segment_ctx) {
    if (!attr_indexer) {
        return Status::InvalidArgs;
    }

    return attr_indexer->BatchRead(segment_ctx);
}

void Segment::DoSelect_(
    const google::protobuf::RepeatedPtrField<std::string>& select_fields,
    SegmentContext* segment_ctx) {
    for (const auto& select_field : select_fields) {
        // check
        int32_t field_id;
        util::ColumnType field_type;
        std::shared_ptr<AttributeIndexer> attr_indexer;
        Status ret = CheckValid_(select_field, &field_id, &field_type, attr_indexer);
        if (ret != Status::OK) {
            continue;
        }

        if (segment_ctx->IsVisitedFieldId(field_id)) {
            continue;
        }

        // get data
        ret = ReadData_(attr_indexer, segment_ctx);
        if (ret != Status::OK) {
            return;
        }
        segment_ctx->InsertVisitedFieldId(field_id);
    }
}

void Segment::DoSelect_(
    DocId doc_id,
    google::protobuf::RepeatedPtrField<util::SelectColumn>* select_fields) {
    indexlib::ArenaWrapper arena_wrapper;
    SegmentContext segment_ctx(arena_wrapper);
    segment_ctx.PushDocId(doc_id);
    DocContext& doc_ctx = (*(segment_ctx.MutableDocContext()))[0];

    for (auto& select_field : *select_fields) {
        // check
        int32_t field_id;
        util::ColumnType field_type;
        std::shared_ptr<AttributeIndexer> attr_indexer;
        Status ret = CheckValid_(
            select_field.column_name(), &field_id, &field_type, attr_indexer);
        if (ret != Status::OK) {
            continue;
        }

        if (segment_ctx.IsVisitedFieldId(field_id)) {
            continue;
        }

        // get data
        ret = ReadData_(attr_indexer, &segment_ctx);
        if (ret != Status::OK) {
            continue;
        }

        util::FieldValue* field_value = doc_ctx.GetField(field_id);
        if (field_value) {
            select_field.mutable_column_value()->Swap(field_value);
        } else {
            if (field_type == util::ColumnType::COLUMN_INT8 ||
                field_type == util::ColumnType::COLUMN_INT16 ||
                field_type == util::ColumnType::COLUMN_INT32 ||
                field_type == util::ColumnType::COLUMN_INT64) {
                auto iter = doc_ctx.single_int_field.find(field_id);
                if (iter != doc_ctx.single_int_field.end()) {
                    select_field.mutable_column_value()->set_int_value(iter->second);
                }
            } else if (field_type == util::ColumnType::COLUMN_FLOAT) {
                auto iter = doc_ctx.single_float_field.find(field_id);
                if (iter != doc_ctx.single_float_field.end()) {
                    select_field.mutable_column_value()->set_double_value(iter->second);
                }
            } else if (field_type == util::ColumnType::COLUMN_DOUBLE) {
                auto iter = doc_ctx.single_double_field.find(field_id);
                if (iter != doc_ctx.single_double_field.end()) {
                    select_field.mutable_column_value()->set_double_value(iter->second);
                }
            }
        }
        
        segment_ctx.InsertVisitedFieldId(field_id);
    }
}

void Segment::DoSelect_(DocId doc_id, int32_t field_id, util::FieldValue* field_value) {
    indexlib::ArenaWrapper arena_wrapper;
    SegmentContext segment_ctx(arena_wrapper);
    segment_ctx.PushDocId(doc_id);
    DocContext& doc_ctx = (*(segment_ctx.MutableDocContext()))[0];

    util::ColumnType field_type;
    std::shared_ptr<AttributeIndexer> attr_indexer;
    Status ret = CheckValid_(field_id, &field_type, attr_indexer);
    if (ret != Status::OK) {
        return;
    }

    // get data
    ret = ReadData_(attr_indexer, &segment_ctx);
    if (ret != Status::OK) {
        return;
    }

    util::FieldValue* doc_field_value = doc_ctx.GetField(field_id);
    if (doc_field_value) {
        field_value->Swap(doc_field_value);
    } else {
        if (field_type == util::ColumnType::COLUMN_INT8 ||
            field_type == util::ColumnType::COLUMN_INT16 ||
            field_type == util::ColumnType::COLUMN_INT32 ||
            field_type == util::ColumnType::COLUMN_INT64) {
            auto iter = doc_ctx.single_int_field.find(field_id);
            if (iter != doc_ctx.single_int_field.end()) {
                field_value->set_int_value(iter->second);
            }
        } else if (field_type == util::ColumnType::COLUMN_FLOAT) {
            auto iter = doc_ctx.single_float_field.find(field_id);
            if (iter != doc_ctx.single_float_field.end()) {
                field_value->set_double_value(iter->second);
            }
        } else if (field_type == util::ColumnType::COLUMN_DOUBLE) {
            auto iter = doc_ctx.single_double_field.find(field_id);
            if (iter != doc_ctx.single_double_field.end()) {
                field_value->set_double_value(iter->second);
            }
        }
    }
}

void Segment::FilterByTTL_(SegmentContext* segment_ctx) {
    if (!TTL()) {
        return;
    }

    int32_t field_id;
    util::ColumnType field_type;
    std::shared_ptr<AttributeIndexer> attr_indexer;
    Status ret = CheckValid_(
        FLAGS_default_ttl_col_name, &field_id, &field_type, attr_indexer);
    if (ret != Status::OK) {
        LOG(WARNING) << "TTL is " << TTL() << " but has not column "
            << FLAGS_default_ttl_col_name;
        return;
    }

    if (unlikely(segment_ctx->IsVisitedFieldId(field_id))) {
        LOG(WARNING) << "TTL column " << FLAGS_default_ttl_col_name
            << " already got and processed in filter_fields or where_fields";
        return;
    }

    ret = ReadData_(attr_indexer, segment_ctx);
    if (ret != Status::OK) {
        LOG(WARNING) << "Get TTL column fail, could not do ttl filter";
        return;
    }

    int64_t current_ts = butil::gettimeofday_s();
    std::vector<DocContext>& doc_ctxs = *(segment_ctx->MutableDocContext());
    for (size_t idx = 0; idx < doc_ctxs.size(); ++idx) {
        DocContext& doc_ctx = doc_ctxs[idx];
        auto iter = doc_ctx.single_int_field.find(field_id);
        if (iter != doc_ctx.single_int_field.end() &&
            iter->second <= current_ts) {
            doc_ctx.filtered = 1;
        }
    }
}

void Segment::DoFilter_(
    const google::protobuf::RepeatedPtrField<util::FilterColumn>& filter_fields,
    const google::protobuf::RepeatedPtrField<util::FilterColumn>& where_fields,
    SegmentContext* segment_ctx) {
    for (const auto& filter_field : filter_fields) {
        if (segment_ctx->RelateDocCount() == 0) {
            break;
        }

        // check
        int32_t field_id;
        util::ColumnType field_type;
        std::shared_ptr<AttributeIndexer> attr_indexer;
        Status ret = CheckValid_(
            filter_field.column_name(), &field_id, &field_type, attr_indexer);
        if (ret != Status::OK) {
            continue;
        }

        if (segment_ctx->IsVisitedFieldId(field_id)) {
            continue;
        }

        // get data
        ret = ReadData_(attr_indexer, segment_ctx);
        if (ret != Status::OK) {
            return;
        }
        segment_ctx->InsertVisitedFieldId(field_id);

        // attribute filter
        attr_indexer->Filter(segment_ctx, filter_field, field_type, field_id);
    }

    for (const auto& where_field : where_fields) {
        if (segment_ctx->RelateDocCount() == 0) {
            break;
        }

        // check
        int32_t field_id;
        util::ColumnType field_type;
        std::shared_ptr<AttributeIndexer> attr_indexer;
        Status ret = CheckValid_(
            where_field.column_name(), &field_id, &field_type, attr_indexer);
        if (ret != Status::OK) {
            continue;
        }

        if (segment_ctx->IsVisitedFieldId(field_id)) {
            continue;
        }

        // get data
        ret = ReadData_(attr_indexer, segment_ctx);
        if (ret != Status::OK) {
            return;
        }
        segment_ctx->InsertVisitedFieldId(field_id);

        // attribute filter
        attr_indexer->Filter(segment_ctx, where_field, field_type, field_id, true);
    }

    FilterByTTL_(segment_ctx);
}

void Segment::DoFilter_(
    const util::FilterColumn& field, SegmentContext* segment_ctx, bool is_where) {
    // check
    int32_t field_id;
    util::ColumnType field_type;
    std::shared_ptr<AttributeIndexer> attr_indexer;
    Status ret = CheckValid_(field.column_name(), &field_id, &field_type, attr_indexer);
    if (ret != Status::OK) {
        return;
    }

    ret = ReadData_(attr_indexer, segment_ctx);
    if (ret != Status::OK) {
        return;
    }

    attr_indexer->Filter(segment_ctx, field, field_type, field_id, is_where);
    FilterByTTL_(segment_ctx);
}

Status Segment::CheckValid_(
    const std::string& field_name,
    int32_t* field_id,
    util::ColumnType* field_type,
    std::shared_ptr<AttributeIndexer>& attr_indexer) const {
    if (!GetFieldIdAndTypedByName(field_name, field_type)) {
        LOG(ERROR) << "field_name[" << field_name << "] Not Found";
        return Status::ConfigError;
    }

    attr_indexer = GetAttributeIndexer(*field_id);
    if (unlikely(!attr_indexer)) {
        LOG(ERROR) << "field_name[" << field_name << "] "
            << " field_id " << *field_id << " index Not Found";
        return Status::NotFound;
    }

    if (unlikely(*field_type != attr_indexer->GetColumnType())) {
        LOG(ERROR) << "request field field_type["
            << static_cast<int32_t>(*field_type) << "] "
            << " not equal attribute index field_type "
            << static_cast<int32_t>(attr_indexer->GetColumnType());
        return Status::ConfigError;
    }

    return Status::OK;
}

Status Segment::CheckValid_(
    int32_t field_id, util::ColumnType* field_type,
    std::shared_ptr<AttributeIndexer>& attr_indexer) const {
    attr_indexer = GetAttributeIndexer(field_id);
    if (unlikely(!attr_indexer)) {
        LOG(ERROR) << "field_id " << field_id << " index Not Found";
        return Status::NotFound;
    }

    *field_type = attr_indexer->GetColumnType();
    return Status::OK;
}

void Segment::TombStoneFilter_(SegmentContext* segment_ctx) const {
    if (!tomb_stone_) {
        return;
    }

    const std::vector<DocId>& doc_ids = segment_ctx->DocIds();
    std::vector<DocContext>& doc_context = *(segment_ctx->MutableDocContext());
    for (size_t idx = 0; idx < doc_ids.size(); ++idx) {
        if (!doc_context[idx].filtered && tomb_stone_->ExistDocId(doc_ids[idx])) {
            doc_context[idx].filtered = 1;
        }
    }
}

void Segment::FilterDeletedDocs_(SegmentContext* segment_ctx) const {
    std::shared_ptr<DeletionMapIndexer> deletion_indexer = GetDeletionIndexer();
    const std::vector<DocId>& doc_ids = segment_ctx->DocIds();
    std::vector<DocContext>& doc_context = *(segment_ctx->MutableDocContext());
    for (size_t idx = 0; idx < doc_ids.size(); ++idx) {
        if (!doc_context[idx].filtered && deletion_index->IsDeleted(doc_ids[idx])) {
            doc_context[idx].filtered = 1;
        }
    }

    TombStoneFilter_(segment_ctx);
}

bool Segment::FilterDeletedDoc_(DocId doc_id) const {
    std::shared_ptr<DeletionMapIndexer> deletion_indexer = GetDeletionIndexer();
    return (deletion_indexer && deletion_indexer->IsDeleted(doc_id))
        || TombStoneFilter_(doc_id);
}

void Segment::FilterAndSelect_(
    SegmentContext* segment_ctx,
    const google::protobuf::RepeatedPtrField<std::string> select_fields,
    const google::protobuf::RepeatedPtrField<util::FilterColumn> filter_fields,
    const google::protobuf::RepeatedPtrField<util::FilterColumn>& where_fields,
    IndexlibRetrieveResponse* resp) {
    // mark-filter deleted docs
    FilterDeletedDocs_(segment_ctx);

    // filter-columns first
    DoFilter_(filter_fields, where_fields, segment_ctx);

    // all is filtered.
    if (segment_ctx->RelateDocCount() == 0) {
        return;
    }

    // then select-columns
    DoSelect_(select_fields, segment_ctx);

    // swap to response as current segment's result
    MoveToResponse_(segment_ctx, select_fields, resp);
}

Status Segment::Query(
    const std::string& invert_field_nane,
    uint64_t term_key,
    uint32_t topk,
    const google::protobuf::RepeatedPtrField<std::string>& select_fields,
    const google::protobuf::RepeatedPtrField<util::FilterColumn>& filter_fields,
    const google::protobuf::RepeatedPtrField<util::FilterColumn>& where_fields,
    IndexlibRetrieveResponse* resp) {
    if (!topk) {
        return Status::OK;
    }

    // get inverted indexer
    int32_t field_id = -1;
    util::ColumnType field_type;
    if (!GetFieldIdAndTypedByName(invert_field_nane, &field_id, &field_type)) {
        return Status::NotFound;
    }
    std::shared_ptr<InvertedIndexer> inverted_indexer = GetInvertedIndexer(field_id);

    // makeup context
    indexlib::ArenaWrapper arena_wrapper;
    SegmentContext segment_ctx(arena_wrapper);
    segment_ctx.Reserve(topk);

    // get doclist by inverted index
    Status ret = inverted_indexer->Lookup(term_key, topk, &segment_ctx);
    if (ret != Status::OK) {
        return ret;
    }
    if (!segment_ctx.RelateDocCount()) {
        return Status::NotFound;
    }

    // attr-filter + attr-select
    FilterAndSelect_(&segment_ctx, select_fields, filter_fields, where_fields, resp);
    return Status::OK;
}

Status Segment::QueryByPk(
    const google::protobuf::RepeatedField<int64_t>& pks,
    const google::protobuf::RepeatedPtrField<std::string>& select_fields,
    const google::protobuf::RepeatedPtrField<util::FilterColumn>& filter_fields,
    const google::protobuf::RepeatedPtrField<util::FilterColumn>& where_fields,
    IndexlibRetrieveResponse* resp) {
    // get pk indexer
    std::shared_ptr<PrimaryKeyIndexer> pk_indexer = GetPkIndexer();

    // makeup context
    indexlib::ArenaWrapper arena_wrapper;
    SegmentContext segment_ctx(arena_wrapper);

    // get doclist by pk indexer
    Status ret = pk_indexer->Lookup(pks, &segment_ctx);
    if (ret != Status::OK) {
        return ret;
    }
    if (!segment_ctx.RelateDocCount()) {
        return Status::OK;
    }

    // mark-filter deleted docs
    FilterDeletedDocs_(&segment_ctx);

    // attr-filter + attr-select
    FilterAndSelect_(&segment_ctx, select_fields, filter_fields, where_fields, resp);
    return Status::OK;
}

ReturnValue<bool> Segment::FilterCheck(
    int64_t pk,
    const google::protobuf::RepeatedPtrField<util::FilterColumn>& filter_fields,
    const google::protobuf::RepeatedPtrField<util::FilterColumn>& where_fields) {
    // get pk indexer
    std::shared_ptr<PrimaryKeyIndexer> pk_indexer = GetPkIndexer();

    ReturnValue<DocId> ret = pk_indexer->Lookup(pk);
    if (ret.Code() != Status::OK) {
        return ReturnValue<bool>{ret.Code(), false};
    }

    if (FilterDeletedDoc_(ret.Value())) {
        return ReturnValue<bool>{Status::Deleted, false};
    }

    indexlib::ArenaWrapper arena_wrapper;
    SegmentContext segment_ctx(arena_wrapper);
    segment_ctx.PushDocId(ret.Value());
    DoFilter_(filter_fields, where_fields, &segment_ctx);
    return ReturnValue<bool>{Status::OK, segment_ctx.RelateDocCount() == 0};
}

ReturnValue<bool> Segment::FilterCheck(
    int64_t pk, const util::FilterColumn& field, bool is_where) {
    // get pk indexer
    std::shared_ptr<PrimaryKeyIndexer> pk_indexer = GetPkIndexer();

    ReturnValue<DocId> ret = pk_indexer->Lookup(pk);
    if (ret.Code() != Status::OK) {
        return ReturnValue<bool>{ret.Code(), false};
    }

    if (FilterDeletedDoc_(ret.Value())) {
        return ReturnValue<bool>{Status::Deleted, false};
    }

    indexlib::ArenaWrapper arena_wrapper;
    SegmentContext segment_ctx(arena_wrapper);
    segment_ctx.PushDocId(ret.Value());
    DoFilter_(field, &segment_ctx, is_where);
    return ReturnValue<bool>{Status::OK, segment_ctx.RelateDocCount() == 0};
}
    
ReturnValue<bool> Segment::FillLocaldata(
    int64_t pk,
    google::protobuf::RepeatedPtrField<util::SelectColumn>* select_fields) {
    // get pk indexer
    std::shared_ptr<PrimaryKeyIndexer> pk_indexer = GetPkIndexer();

    ReturnValue<DocId> ret = pk_indexer->Lookup(pk);
    if (ret.Code() != Status::OK) {
        return ReturnValue<bool>{ret.Code(), 0};
    }

    if (FilterDeletedDoc_(ret.Value())) {
        return ReturnValue<bool>{Status::Deleted, 0};
    }

    DoSelect_(ret.Value(), select_fields);
    return ReturnValue<bool>{Status::OK, true};
}

ReturnValue<bool> Segment::FillLocaldata(int64_t pk, util::SelectColumn* select_field) {
    // get pk indexer
    std::shared_ptr<PrimaryKeyIndexer> pk_indexer = GetPkIndexer();

    ReturnValue<DocId> ret = pk_indexer->Lookup(pk);
    if (ret.Code() != Status::OK) {
        return ReturnValue<bool>{ret.Code(), 0};
    }

    if (FilterDeletedDoc_(ret.Value())) {
        return ReturnValue<bool>{Status::Deleted, 0};
    }

    google::protobuf::RepeatedPtrField<util::SelectColumn> select_fields;
    util::SelectColumn* col = select_fields.Add();
    col->Swap(select_field);
    DoSelect_(ret.Value(), &select_fields);
    select_field->Swap(col);
    return ReturnValue<bool>{Status::OK, true};
}

ReturnValue<bool> Segment::FillLocaldata(
    int64_t pk, int32_t field_id, util::FieldValue* field_value) {
    // get pk indexer
    std::shared_ptr<PrimaryKeyIndexer> pk_indexer = GetPkIndexer();

    ReturnValue<DocId> ret = pk_indexer->Lookup(pk);
    if (ret.Code() != Status::OK) {
        return ReturnValue<bool>{ret.Code(), 0};
    }

    if (FilterDeletedDoc_(ret.Value())) {
        return ReturnValue<bool>{Status::Deleted, 0};
    }

    DoSelect_(ret.Value(), field_id, field_value);
    return ReturnValue<bool>{Status::OK, true};
}

void Segment::CheckInvertedFieldModified_(
    const IndexlibMessage& message,
    std::vector<std::pair<int32_t, const util::FieldValue&>>* inverted_fields) const {
    for (const auto& column : message.columns()) {
        const std::string& field_name = column.column_name();
        const util::FieldValue& field_value = column.column_value();

        int32_t field_id = -1;
        util::ColumnType field_type;
        GetFieldIdAndTypedByName(field_name, &field_id, &field_type);
        auto inverted_index = GetInvertedIndexer(field_id);
        if (inverted_index) {
            inverted_fields->emplace_back(field_id, field_value);
        }
    }
}

bool Segment::CalcTerm_(
    int32_t field_id, const SegmentContext& segment_ctx, uint64_t* term) const {
    const auto& field_config = Schema()->fields(field_id);
    switch (field_config.field_type()) {
    case util::COLUMN_INT8:
    case util::COLUMN_INT16:
    case util::COLUMN_INT32:
    case util::COLUMN_INT64: {
        int64_t value;
        segment_ctx.DocContexts[0].GetField(field_id, &value);
        GetFieldHash(value, term):
    }
    break;
    case util::COLUMN_FLOAT: {
        float value;
        segment_ctx.DocContexts[0].GetField(field_id, &value);
        GetFieldHash(value, term):
    }
    case util::COLUMN_DOUBLE: {
        double value;
        segment_ctx.DocContexts[0].GetField(field_id, &value);
        GetFieldHash(value, term):
    }
    break;
    case util::COLUMN_STRING: {
        util::FieldValue* field_value = segment_ctx.DocContexts[0].GetField(field_id);
        GetFieldHash(field_config, field_value, term);
    }
    break;
    default:
        return false;
    }

    return true;
}

util::HawkingRecallRetCode Segment::UpdateInvertedIndex_(
    const std::unordered_set<int32_t>& exclude_field_ids,
    const IndexlibMessage& message,
    DocId doc_id) {
    // check which field_ids is need to modify inverted index
    std::vector<std::pair<int32_t, const util::FieldValue&>> inverted_field_ids;
    CheckInvertedFieldModified_(&inverted_field_ids);
    if (inverted_field_ids.empty()) {
        return util::HawkingRecallRetCode::OK;
    }

    // traverse every field of inverted index, calc old_term and new_term,
    // then remove doc_id from old_term, add doc_id from new_term
    for (auto field : inverted_field_ids) {
        if (exclude_field_ids.find(field.first) != exclude_field_ids.end()) {
            continue;
        }

        uint64_t old_term, new_term;
        if (unlikely(!CalcTerm_(field.first, segment_ctx, &old_term))) {
            LOG(ERROR) << "calc current term fail with field_id "
                << field.first << " from attribute index of doc_id "
                << doc_id << " fail";
            return util::HawkingRecallRetCode::INDEX_CONFIG_INVALID;
        }

        const FieldConfig& field_config = Schema()->fields(field.first);
        if (unlikely(!GetFieldHash(field_config, &field.second, &new_term))) {
            LOG(ERROR) << "calc new term fail with field_id "
                << field.first << " from attribute index of doc_id "
                << doc_id << " fail";
            return util::HawkingRecallRetCode::INDEX_CONFIG_INVALID;
        }

        auto inverted_index = GetInvertedIndexer(field_id);
        inverted_index->UpdateDocument(old_term, new_term, doc_id);
    }
}

bool Segment::CheckIsSameFieldValue_(
    const SegmentContext& segment_ctx, int32_t field_id,
    util::ColumnType field_type, const FieldValue& field_value) const {
    switch (field_type) {
    case util::ColumnType::COLUMN_INT8:
    case util::ColumnType::COLUMN_INT16:
    case util::ColumnType::COLUMN_INT32:
    case util::ColumnType::COLUMN_INT64: {
        int64_t cur_val;
        segment_ctx.DocContexts[0].GetField(field_id, &cur_val);
        return cur_val == field_value.int_value();
    }
    break;
    case util::ColumnType::COLUMN_FLOAT: {
        float cur_val;
        segment_ctx.DocContexts[0].GetField(field_id, &cur_val);
        return cur_val == static_cast<float>(field_value.double_value());
    }
    break;
    case util::ColumnType::COLUMN_DOUBLE: {
        double cur_val;
        segment_ctx.DocContexts[0].GetField(field_id, &cur_val);
        return cur_val == field_value.double_value();
    }
    default: {
        util::FieldValue* cur_val = segment_ctx.DocContexts[0].GetField(field_id);
        return google::protobuf::util::MessageDifferencer::Equals(*cur_val, field_value);
    }
    }

    return true;
}

util::HawkingRecallRetCode Segment::ExcludeFakeModify_(
    const IndexlibMessage& message, DocId doc_id,
    SegmentContext* segment_ctx,
    std::unordered_set<int32_t>* exclude_field_ids,
    bool* with_diff) const {
    
    for (const auto& column : message.columns()) {
        const std::string& field_name = column.column_name();
        const FieldValue& field_value = column.column_value();
        int32_t field_id = -1;
        util::ColumnType field_type;
        if (!GetFieldIdAndTypedByName(field_name, &field_id, &field_type)) {
            LOG(WARNING) << "unknown field_name when update: " << field_name;
            continue;
        }
        
        auto attribute_index = GetAttributeIndexer(field_id);
        if (Status::OK != attribute_index->BatchRead(&segment_ctx)) {
            LOG(ERROR) << "get current field_value of field_id "
                << field_id << " from attribute index of doc_id "
                << doc_id << " fail";
            return util::HawkingRecallRetCode::INDEX_SEARCH_NULL;
        }

        if (CheckIsSameFieldValue_(*segment_ctx, field_id, field_type, field_value)) {
            exclude_field_ids->emplace(field_id);
            LOG(WARNING) << "without modify with field_name " << field_name;
        } else {
            *with_diff = true;
        }
    }

    return util::HawkingRecallRetCode::OK;
}

util::HawkingRecallRetCode Segment::UpdateDocument_(
    const IndexlibMessage& message, uint64_t pk,
    std::shared_ptr<NormalDocument> doc, DocId doc_id) {
    // 1. same field_value with current field_value, exclude
    indexlib::ArenaWrapper arena_wrapper;
    SegmentContext segment_ctx(arena_wrapper);
    segment_ctx.PushDocId(doc_id);

    bool with_diff = false;
    std::unordered_set<int32_t> exclude_field_ids;
    auto ret =
        ExcludeFakeModify_(message, doc_id, &segment_ctx, &exclude_field_ids, &with_diff);
    if (util::HawkingRecallRetCode::OK != ret)
        return ret;
    }
    if (!with_diff) {
        LOG(WARNING) << "Update valuse are all same as current";
        return util::HawkingRecallRetCode::OK;
    }
    
    // 2. inverted update
    util::HawkingRecallRetCode ret =
        UpdateInvertedIndex_(exclude_field_ids, message， doc_id);
    if (ret != util::HawkingRecallRetCode::OK) {
        return ret;
    }

    return UpdateAttributeIndex_(exclude_field_ids);
}

}
}